package org.util.concurrent;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author liangsu
 * @version v1.0
 * @Description
 * @Date 2020/11/6 9:16
 * @since
 */
public class BindThreadExecutor2 implements Executor {
//    private final Logger logger = LoggerFactory.getLogger(BindThreadExecutor2.class);

    /** 核心线程数 */
    private int coreSize;

    /** 工作线程的队列创建工厂 */
    private BindThreadQueueFactory queueFactory;
    /** 线程工厂 */
    private ThreadFactory threadFactory;

    private RejectedExecutionHandler handler;

    /** 工作线程 */
    private Thread[] workers;
    /** 工作线程的队列 */
    private BlockingQueue<Runnable>[] blockingQueues;

    private AtomicInteger threadSelector = new AtomicInteger(0);

    public BindThreadExecutor2(int coreSize){
        this(coreSize, Executors.defaultThreadFactory(), BindThreadQueueFactory.DEFAULT);
    }

    /**
     *
     * @param coreSize
     * @param threadFactory
     * @param queueFactory
     */
    public BindThreadExecutor2(int coreSize, ThreadFactory threadFactory, BindThreadQueueFactory queueFactory){
        this.coreSize = coreSize;
        this.workers = new Thread[coreSize];
        this.threadFactory = threadFactory;
        this.queueFactory = queueFactory;
    }

    /**
     * 提交任务
     * @param command
     */
    @Override
    public void execute(Runnable command) {
        execute(command, threadSelector.getAndIncrement());
    }

    /**
     * 将任务绑定到指定的线程上执行
     * @param command
     * @param threadId 线程id，用于决定在那个线程上执行任务
     */
    public void execute(Runnable command, int threadId) {
        if (command == null)
            throw new NullPointerException();

        if(threadId < 0){
            throw new IllegalArgumentException("thread id exception: " + threadId);
        }
        // 计算在哪个线程上执行
        int index = threadId % workers.length;

        // 获取任务队列，为空则创建
        if(blockingQueues[index] == null){
            synchronized (blockingQueues){
                if(blockingQueues[index] == null){
                    blockingQueues[index] = queueFactory.newQueue();
                }
            }
        }

        // 提交任务
        boolean addTask = blockingQueues[index].offer(command);
        if(!addTask){ // 添加任务失败

        }

        // 创建线程
        if(workers[index] == null){
            synchronized (workers){
                if(workers[index] == null){
                    BindThreadWorker worker = new BindThreadWorker(blockingQueues[index]);
                    workers[index] = threadFactory.newThread(worker);
                    workers[index].start();
                }
            }
        }
    }

    /**
     * 执行任务的线程
     */
    class BindThreadWorker implements Runnable{

        BlockingQueue<Runnable> queue;

        public BindThreadWorker(BlockingQueue<Runnable> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            try{
                Runnable task = null;
                while (true){
                    task = queue.take();
                    if(task != null){
                        task.run();
                    }
                }
            }catch (Exception e){
//                logger.error("execute task error", e);

            }
        }
    }

}
